package com.py.custom.ability.kafka.producer.base;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * 生产者默认实现
 *
 * @param <K>
 * @param <V>
 */
public class DefaultProducerProcessor<K, V> extends AbstractProducerProcessor<K, V> {

    private final Properties properties;

    private final String topicName;

    private final Producer<K, V> producer;

    public DefaultProducerProcessor(String topicName, Properties properties) {
        this.properties = properties;
        this.topicName = topicName;
        this.producer = new KafkaProducer<K, V>(properties);
    }

    @Override
    public ProducerResult sendMessage(V message) throws Exception {
        System.out.println("DefaultProducerProcessor [ 普通消息发送 ]");
        return sendMessage(topicName, message);
    }

    @Override
    public ProducerResult sendMessage(String topicName, V message) throws Exception {
        ProducerRecord<K, V> record = new ProducerRecord<>(topicName, message);
        return doSend(record);
    }

    private ProducerResult doSend(ProducerRecord<K, V> record) {
        Future<RecordMetadata> metadataFuture = producer.send(record);
        try {
            RecordMetadata metadata = metadataFuture.get();
        } catch (InterruptedException | ExecutionException e) {
            // 异常处理
        }
        // 返回结果处理
        return null;
    }

    @Override
    public void close() throws Exception {
        // 关闭生产者回收资源
    }
}
